package flink;

import flink.model.CountWithTimestamp;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;

public class CountWithTimeoutFunction extends KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Long>> {

    // 使用 ValueState
    private ValueState<CountWithTimestamp> state;

    @Override
    public void open(Configuration parameters) throws Exception {

        // 初始化状态，name是myState
        state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
    }

    @Override
    public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {
        // 取得当前是哪个单词
        Tuple currentKey = ctx.getCurrentKey();

        // 从backend取得当前单词的myState状态
        CountWithTimestamp current = state.value();

        // 如果myState还从未没有赋值过，就在此初始化
        if (current == null) {
            current = new CountWithTimestamp();
            current.key = value.f0;
        }

        // 单词数量加一
        current.count++;

        // 取当前元素的时间戳，作为该单词最后一次出现的时间
        current.lastModified = ctx.timestamp();

        // 重新保存到backend，包括该单词出现的次数，以及最后一次出现的时间
        state.update(current);

        // 为当前单词创建定时器，十秒后后触发
        long timer = current.lastModified + 10000;

        ctx.timerService().registerProcessingTimeTimer(timer);

        // 打印所有信息，用于核对数据正确性
        System.out.println(String.format("process, %s, %d, lastModified : %d (%s), timer : %d (%s)\n\n",
                currentKey.getField(0),
                current.count,
                current.lastModified,
                transTime(current.lastModified),
                timer,
                transTime(timer)));

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out) throws Exception {
        // 取得当前单词
        Tuple currentKey = ctx.getCurrentKey();

        // 取得该单词的myState状态
        CountWithTimestamp result = state.value();

        // 当前元素是否已经连续10秒未出现的标志
        boolean isTimeout = false;

        // timestamp是定时器触发时间，如果等于最后一次更新时间+10秒，就表示这十秒内已经收到过该单词了，
        // 这种连续十秒没有出现的元素，被发送到下游算子
        if (timestamp == result.lastModified + 10000) {
            // 发送
            out.collect(new Tuple2<String, Long>(result.key, result.count));

            isTimeout = true;
        }

        // 打印数据，用于核对是否符合预期
        System.out.println(String.format("ontimer, %s, %d, lastModified : %d (%s), stamp : %d (%s), isTimeout : %s\n\n",
                currentKey.getField(0),
                result.count,
                result.lastModified,
                transTime(result.lastModified),
                timestamp,
                transTime(timestamp),
                String.valueOf(isTimeout)));
    }


    public static String transTime(long timeStamp) {
        return new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date(timeStamp));
    }

}

